package day3;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseSink extends RichSinkFunction<String[]> {

    Connection connection;

    Table table;

    public HbaseSink() {
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum" , "bigdata1,bigdata2,bigdata3");
        conf.set("hbase.zookeeper.property.clientPort" , "2181");
        connection = ConnectionFactory.createConnection();
        table = connection.getTable(TableName.valueOf(Bytes.toBytes("shtd_result"), Bytes.toBytes("order_info")));
    }

    @Override
    public void invoke(String[] value, Context context) throws Exception {
        Put put = new Put(Bytes.toBytes(value[0]));
        for (int i = 0; i < value.length; i++) {
            put.addColumn(Bytes.toBytes("info") , Bytes.toBytes(value[i]) , Bytes.toBytes(value[i]));
        }
        table.put(put);
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
        if (table != null){
            table.close();
        }
    }
}
